package com.example.action;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.repository.CrudRepository;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import com.example.dao.BwRyBjtry;

@Component
@ConditionalOnProperty(prefix = "spring.profiles", name = "active", havingValue = "kafka_manal", matchIfMissing = false)
public class KafkaConsumer2 extends BaseKafkaConsumer {

	@Autowired
	CrudRepository<BwRyBjtry, String> bwRyBjtryDAO;
	
	//@Autowired
	//CrudRepository<BwXyBjxy, String> bwxtBjtxyDAO;

	@KafkaListener(topics = "BW_RY_BJTRY")
	public void bwRyBjtryConsumer(String message, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key,
			@Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Acknowledgment ack) {
		consumer(message, key, topic, ack, bwRyBjtryDAO, BwRyBjtry::new);
	}
	
	//@KafkaListener(topics = "BW_XY_BJTXY")
	//public void bwRyBjtryConsumer(String message, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key,
	//		@Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Acknowledgment ack) {
	//	consumer(message, key, topic, ack, bwxtBjtxyDAO, BwXyBjtxy::new);
	//}
}

